package com.wp.kafka.consumer;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;

import java.util.Collections;
import java.util.Map;
import java.util.Properties;
import java.util.Set;

public class KafkaConsumerWithOffsetControl1 {
    public static void main(String[] args) {
        Properties prop = new Properties();
        //bootstrap-server
        prop.put("bootstrap.servers", "192.168.100.141:9092");
        // 指定消费者组
        prop.put("group.id", "wp1");
        // 指定消费位置: earliest/latest/none
        prop.put("auto.offset.reset", "earliest");
        // 指定消费的key的反序列化方式
        prop.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        // 指定消费的value的反序列化方式
        prop.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        prop.put("enable.auto.commit", "false");
        prop.put("session.timeout.ms", "60000");
        prop.put("max.poll.interval.ms", "60000");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(prop);
        consumer.subscribe(Collections.singletonList("chantree-all"));

        /*
        先消费一下数据，这时候会为消费者分配partition，此时才能用consumer.assignment();获取到分配的partition
         */
        consumer.poll(10000);
        Set<TopicPartition> assignmentset = consumer.assignment();

        Map<TopicPartition, Long> topicPartitionoffsetsMap = consumer.beginningOffsets(assignmentset);
        Set<Map.Entry<TopicPartition, Long>> entrySet = topicPartitionoffsetsMap.entrySet();
        for (Map.Entry<TopicPartition, Long> entry : entrySet) {
            System.out.println("topicpartion=" + entry.getKey() + ",beginningOffset=" + entry.getValue());
        }

        Map<TopicPartition, Long> topicPartitionEndOffsets = consumer.endOffsets(assignmentset);
        entrySet = topicPartitionEndOffsets.entrySet();
        for (Map.Entry<TopicPartition, Long> entry : entrySet) {
            System.out.println("topicpartion=" + entry.getKey() + ",endOffset=" + entry.getValue());
        }

        //重置offset到开始
        consumer.seekToBeginning(assignmentset);
        for (int i = 0; i < 10; i++) {
            ConsumerRecords<String, String> records = consumer.poll(10000);
            if (records.isEmpty()) {
                break;
            }
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
            }
            try {
                //同步提交，阻塞方法
                consumer.commitSync();
            } catch (Exception e) {
                System.out.println("commit failed");
            }
        }
        topicPartitionoffsetsMap = consumer.beginningOffsets(assignmentset);
        entrySet = topicPartitionoffsetsMap.entrySet();
        for (Map.Entry<TopicPartition, Long> entry : entrySet) {
            System.out.println("topicpartion=" + entry.getKey() + ",beginningOffset=" + entry.getValue());
        }

        topicPartitionEndOffsets = consumer.endOffsets(assignmentset);
        entrySet = topicPartitionEndOffsets.entrySet();
        for (Map.Entry<TopicPartition, Long> entry : entrySet) {
            System.out.println("topicpartion=" + entry.getKey() + ",endOffset=" + entry.getValue());
        }
        consumer.close();
    }
}
